package org.fastsyncer.worker.framework;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.parser.framework.ParserFactory;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * <h3>Java 线程池 Java通过Executors提供四种线程池，分别为：</h3>
 * <pre>
 * newCachedThreadPool创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
 * newFixedThreadPool 创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。 newScheduledThreadPool
 * 创建一个定长线程池，支持定时及周期性任务执行。 newSingleThreadExecutor
 * 创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
 * </pre>
 * @author AE86
 * @date 2017年5月13日 下午7:16:37
 * @version 1.0.0
 */
public final class WorkerFactory {

    private static final Logger logger = LoggerFactory.getLogger(WorkerFactory.class);

    private WorkerFactory(){
    }
    
    private static volatile WorkerFactory instance = null;

    public static WorkerFactory getInstance() {
        if (null == instance) {
            synchronized (WorkerFactory.class) {
                if (null == instance) {
                    instance = new WorkerFactory();
                }
            }
        }
        return instance;
    }

    private ParserFactory parser = ParserFactory.getInstance();
    
    /**
     * 全量同步
     * 
     * @param mappingTask
     * @param distributeTask 存储所有Task,每个Task里面包含执行的任务
     * @return
     * @throws JSONException 
     * @throws InterruptedException 
     */
    public JSONObject handle(MappingTask mappingTask, Queue<Integer[]> distributeTask) throws InterruptedException, JSONException {
        logger.info("Task:" + mappingTask.getId());
        // 获取配置的线程个数
        Integer threadNum = mappingTask.getThreadNum();
        // 当前任务个数
        int tSize = distributeTask.size();
        // 为了节省不必要的线程资源，当线程个数超过实际处理的任务，那么只需要开辟处理的任务个数即可
        threadNum = threadNum > tSize ? tSize : threadNum;
        // CountDownLatch，一个同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。
        CountDownLatch latch = new CountDownLatch(threadNum);
        // 创建线程池
        ExecutorService es = Executors.newCachedThreadPool();
        // 错误信息
        Queue<JSONObject> reasonQuene = new ConcurrentLinkedQueue<>();
        AtomicInteger total = new AtomicInteger(0);
        AtomicInteger success = new AtomicInteger(0);
        // 分配配置中设置的线程数
        for (int i = 0; i < threadNum; i++) {
            es.execute(new Runnable() {

                @Override
                public void run() {
                    while (!distributeTask.isEmpty()) {
                        // 消费
                        Integer[] pos = distributeTask.poll();
                        if(pos==null || pos.length != 2){
                            break;
                        }
                        try {
                            Integer[] result = parser.parse(mappingTask, pos, reasonQuene);
                            // 累加执行总数、成功条数
                            total.getAndAdd(result[0]);
                            success.getAndAdd(result[1]);
                        } catch (Exception e) {
                            logger.error(e.getMessage());
                        }
                    }
                    latch.countDown();
                }

            });
        }
        // 使得主线程(main)阻塞直到latch.countDown()为零才继续执行
        latch.await();
        es.shutdown();

        // 统计错误原因，因为考虑产生的错误不会很多，size不会影响性能
        JSONArray err = new JSONArray();
        int i = 0;
        while (!reasonQuene.isEmpty()) {
            JSONObject e = reasonQuene.poll();
            err.put(i, e);
            logger.info("Record error(" + (i + 1) + "):" + e);
            i++;
        }
        // 统计执行结果
        return handleResult(mappingTask.getId(), success.get(), (total.get() - success.get()), err);
    }

    /**
     * 增量同步
     * 
     * @param mappingTask
     * @param rows
     * @return
     * @throws JSONException
     */
    public JSONObject handle(MappingTask mappingTask, JSONArray rows) throws JSONException {
        int fail = 0;
        int total = rows.length();
        JSONArray err = new JSONArray();
        for (int i = 0; i < total; i++) {
            JSONObject row = rows.getJSONObject(i);
            try {
                // 执行增量操作，如果失败会抛出异常，否则返回true
                parser.parse(mappingTask, row);
            } catch (Exception e) {
                fail++;
                handleException(e, err, row);
            }
        }
        return handleResult(mappingTask.getId(), (total - fail), fail, err);
    }

    /**
     * 统一响应结果
     * @param driverId
     * @param s 成功数
     * @param f 失败数
     * @param err 错误消息
     * @return
     * @throws JSONException
     */
    private JSONObject handleResult(String driverId, int s, int f, JSONArray err) throws JSONException{
        JSONObject r = new JSONObject();
        r.put("id", driverId);
        r.put("success", s);
        r.put("fail", f);
        r.put("reason", err);
        return r;
    }
    
    private void handleException(Exception e, JSONArray result, JSONObject row) {
        try {
            // 记录错误信息
            JSONObject err = new JSONObject();
            err.put("params", row);
            err.put("message", e.getMessage());
            result.put(err);
        } catch (JSONException ex) {
            logger.error("记录错误信息时：转换JSONObject异常！", ex);
        }
    }

}
